/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package localmq

import (
	"sync"
	"time"

	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/types"
	drivers "hundsun.com/hsl/hschain/store/mq"
)

const (
	// MqName ...
	MqName = "localmq"
)

var once sync.Once
var q Queue

// Service localmq service
type Service struct {
	Client Client
}

func init() {
	drivers.Register(MqName, New)
}

// New 新建实例 mode:worker(task)模式, topic模式
func New(_ *config.PluginConfig, mode string) (drivers.Client, error) {
	once.Do(func() {
		// 全局唯一queue 并且start
		q = NewQueue(DefaultQueueName)
		go func(q *Queue) {
			(*q).Start()
		}(&q)
	})
	s := &Service{}
	s.Client = q.Client(mode)
	return s, nil
}

// Send 发送消息
func (s *Service) Send(msg *drivers.Message, sync bool) error {
	m, err := s.msg(msg)
	if err != nil {
		return err
	}
	return s.Client.Send(m, sync)
}

// SendTimeout 带超时的消息发送
func (s *Service) SendTimeout(msg *drivers.Message, sync bool, timeout time.Duration) error {
	m, err := s.msg(msg)
	if err != nil {
		return err
	}
	return s.Client.SendTimeout(m, sync, timeout)
}

// Wait 等待应答消息(req--rsp)
func (s *Service) Wait(msg *drivers.Message) (*drivers.Message, error) {
	m, err := s.msg(msg)
	if err != nil {
		return nil, err
	}

	m, err = s.Client.Wait(m)
	if err != nil {
		return nil, err
	}
	return s.toMsg(m), nil
}

// WaitTimeout 等待应答消息(req--rsp)
func (s *Service) WaitTimeout(msg *drivers.Message, timeout time.Duration) (*drivers.Message, error) {
	m, err := s.msg(msg)
	if err != nil {
		return nil, err
	}

	m, err = s.Client.WaitTimeout(m, timeout)
	if err != nil {
		return nil, err
	}
	return s.toMsg(m), nil
}

// Recv 接收订阅消息
func (s *Service) Recv() chan *drivers.Message {
	// buffer with the same capacity 5
	c := make(chan *drivers.Message, 5)
	go func() {
		for {
			m, isOpen := <-s.Client.Recv()
			if !isOpen {
				close(c)
				for msg := range c {
					replyMsg := s.NewMessage("", "", msg.Ty, types.ErrChannelClosed)
					msg.Reply(replyMsg)
				}
				return
			}

			if m != nil {
				c <- s.toMsg(m)
			}
		}
	}()
	return c
}

// Sub 订阅消息
func (s *Service) Sub(mainTopic, subTopic string) error {
	return s.Client.Sub(mainTopic, subTopic)
}

// Close 关闭client
func (s *Service) Close() {
	s.Client.Close()
}

// NewMessage 新建消息
func (s *Service) NewMessage(mainTopic, subTopic string, ty int64, data interface{}) (msg *drivers.Message) {
	topic, _ := genTopicKey(mainTopic, subTopic)
	m := s.Client.NewMessage(topic, ty, data)
	msg = s.toMsg(m)
	if msg.ChReply == nil {
		msg.ChReply = make(chan *drivers.Message, 1)
	}
	m.SetMsgChan(msg.ChReply)
	return msg
}

func (s *Service) msg(msg *drivers.Message) (*Message, error) {
	topic, err := genTopicKey(msg.MainTopic, msg.SubTopic)
	if err != nil {
		return nil, err
	}

	return &Message{Topic: topic, Ty: msg.Ty, Data: msg.Data, chMsgReply: msg.ChReply}, nil
}

func (s *Service) toMsg(m *Message) *drivers.Message {
	mainTopic, subTopic := splitTopicKey(m.Topic)
	return &drivers.Message{MainTopic: mainTopic, SubTopic: subTopic, Ty: m.Ty, Data: m.Data, ChReply: m.chMsgReply, Err: m.Err()}
}
